package com.haoziqi.chapter_07;

import com.sun.prism.shader.AlphaOne_LinearGradient_AlphaTest_Loader;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * description
 * created by A on 2021/3/11
 */
public class Flink04_Trigger {
    public static void main(String[] args) {
        //1.创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> DS = env.socketTextStream("hadoop102", 9999);
        SingleOutputStreamOperator<WaterSensor> DM = DS.map(
                new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        String[] split = value.split(",");

                        return new WaterSensor(
                                split[0], Long.valueOf(split[1]), Integer.valueOf(split[2])
                        );
                    }
                }
        );
        DM.keyBy(new KeySelector<WaterSensor, String>() {
            @Override
            public String getKey(WaterSensor value) throws Exception {
                return value.getId();
            }
        });
        KeyedStream<WaterSensor, String> KS = DM.keyBy(value -> value.getId());
        SingleOutputStreamOperator<String> pro = KS.window(
                SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(2))
        ).trigger(
                new Trigger<WaterSensor, TimeWindow>() {
                    /**
                     * 每来一条数据,处理一次该方法
                     * @param element
                     * @param timestamp
                     * @param window
                     * @param ctx
                     * @return
                     * @throws Exception
                     */
                    @Override
                    public TriggerResult onElement(WaterSensor element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
                        return TriggerResult.FIRE_AND_PURGE;
                    }

                    //处理时间，达到某个时间点，执行该方法 ： 触发条件： 处理时间到了预设的时间
                    @Override
                    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
                        return null;
                    }

                    /**
                     * 事件时间，达到某个时间点，执行该方法： 触发条件： WaterMark到了预设的时间
                     * @param time
                     * @param window
                     * @param ctx
                     * @return
                     * @throws Exception
                     */
                    @Override
                    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
                        return null;
                    }

                    @Override
                    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {

                    }
                }
        ).process(
                //输入类型，输出类型，key的类型，窗口类型
                new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
                    @Override
                    public void process(String key, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                        out.collect("key=" + key + "\n" +
                                "数据条数:" + elements.spliterator().estimateSize() + "\n" +
                                "窗口是:[" + context.window().getStart() + "," + context.window().getEnd() + ")\n\n");
                    }
                }
        );
        pro.print();

        //提交任务
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }



    }
}
